import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ValueStateDemo {

    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        //2.设置模式和并行数
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        //3.指定source为socket
        DataStream<String> socketDs = env.socketTextStream("192.168.123.8",8888);
        //4.进行map转换为Tuple3， 源数据为:iphone12,2022-11-11 10:00:00,2399 的数据
        DataStream<Tuple3<String, String, Integer>> map = socketDs.map(new MapFunction<String, Tuple3<String, String, Integer>>() {

            @Override
            public Tuple3<String, String, Integer> map(String value) throws Exception {
                String[] strArr = value.split(",");
                return Tuple3.of(strArr[0],strArr[1],Integer.parseInt(strArr[2]));
            }
        });
        //5.按商品名称进行分组
        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = map.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        });
        //6.进行RichMapFunction ,自定义ValueState实现maxby，输出itemName,最大金额的 元组
        DataStream<Tuple2<String, Integer>> maxDS = keyedStream.map(new RichMapFunction<Tuple3<String, String, Integer>, Tuple2<String, Integer>>() {

            /**
             * valuestate暂存对象，需要统计价格，因此泛型为Integer
             */
            private ValueState<Integer> valueState = null;
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //对valuestate初始化
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("maxValue",Integer.class));
            }
            @Override
            public void close() throws Exception {
                super.close();
            }
            @Override
            public Tuple2<String, Integer> map(Tuple3<String, String, Integer> order) throws Exception {
                //先取出历史值
                Integer maxValue = valueState.value();
                //首次没有历史值，或是之后大于历史值，则做更新
                if(maxValue ==null || order.f2 >maxValue){
                    valueState.update(order.f2);
                }
                return Tuple2.of(order.f0,valueState.value());
            }
        });
        maxDS.print("当前最大");

        try {
            env.execute("ValueState demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
